package insurance.udf

import org.apache.spark.sql.expressions.{MutableAggregationBuffer, UserDefinedAggregateFunction}
import org.apache.spark.sql.types.{DataType, DoubleType, IntegerType, StructField, StructType}
import org.apache.spark.sql.{Row, SparkSession}

import java.util.concurrent.TimeUnit

object udaf1 {
  object CalcLx extends UserDefinedAggregateFunction {
    // 聚合函数的输入数据结构
    override def inputSchema: StructType = new StructType()
      .add(StructField("qx", DoubleType))
      .add(StructField("lx", DoubleType))
      .add(StructField("bpp_",IntegerType))
    // 缓存区数据结构
    override def bufferSchema: StructType = new StructType()
      .add(StructField("buffer_qx", DoubleType))
      .add(StructField("buffer_lx", DoubleType))
    // 聚合函数返回值数据结构
    override def dataType: DataType = DoubleType
    // 聚合函数是否是幂等的，即相同输入是否总是能得到相同输出
    override def deterministic: Boolean = true
    // 初始化缓冲区
    override def initialize(buffer: MutableAggregationBuffer): Unit = {
      buffer(0) = null
      buffer(1) = null
    }
    // 分组后的数据，会一行一行地依次进入这个update方法进行处理
    override def update(buffer: MutableAggregationBuffer, input: Row): Unit = {
      //input.get(1)就是本轮次的lx字段。
      if (null != input.get(1)) {
        //只有当policy_year=1时，才会进来。
        //将第一轮次的qx,lx直接分别更新到缓存中。
        buffer(0) = input.getDouble(0)
        buffer(1) = input.getDouble(1)
      } else {
        //只有当policy_year>1时，才会进来。
        val thisqx = input.getDouble(0)
        //从缓存中拿出上一轮次的qx
        val lastqx = buffer.getDouble(0)
        //从缓存中拿出上一轮次的lx
        val lastlx = buffer.getDouble(1)
        //当前数据的bpp_
        val thisbpp_ = input.getInt(2)
        //本轮次的最新qx值，更新进缓存。给下轮次使用
        buffer(0) = thisqx
        //本轮次的最新lx值，更新进缓存。给下轮次使用
        val thislx = lastlx * (1 - lastqx) * thisbpp_
        buffer(1) = thislx
      }
    }
    // 如果有多个分区，那么每个分区都会维护一个缓冲区，分区间如何合并缓冲区成为一个缓冲区
    // 合并聚合函数缓冲区，如果用作窗口函数且有order by排序的话，只会有一个分区，不用考虑合并，方法体直接给个null。
    override def merge(buffer1: MutableAggregationBuffer, buffer2: Row): Unit = {

    }
    // 计算最终结果
    override def evaluate(buffer: Row): Any = {
      //把缓存的最新lx值取出来。
      buffer.getDouble(1)
    }
  }
}
